package com.atguigu.chapter08;

import com.atguigu.chapter02.Event;
import com.atguigu.chapter05.Clicksource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import scala.Tuple3;

import java.time.Duration;

public class SplitStreamByOutputTag {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new Clicksource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        }));

        OutputTag<Tuple3<String, String, String>> MaryTag = new OutputTag<Tuple3<String, String, String>>("Mary-outputtag") {};
        OutputTag<Tuple3<String, String, String>> BobTag = new OutputTag<Tuple3<String, String, String>>("BobTag-outputtag") {};
        SingleOutputStreamOperator<Event> processStream = stream.process(new ProcessFunction<Event, Event>() {
            @Override
            public void processElement(Event event, Context context, Collector<Event> collector) throws Exception {
                if (event.user.equals("Mary"))
                    context.output(MaryTag, new Tuple3(event.user, event.url, event.timestamp.toString()));
                else if (event.user.equals("Bob"))
                    context.output(BobTag, new Tuple3(event.user, event.url, event.timestamp.toString()));
                else
                    collector.collect(event);
            }
        });
        processStream.print("else");
        processStream.getSideOutput(MaryTag).print("Mary");
        processStream.getSideOutput(BobTag).print("Bob");

        env.execute();

    }
}
